/*
 * Copyright (c) 2024 endless-sky
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "pcm_aac_encoder.h"

#include "utils_log.h"

#include <cstdio>
#include <cstring>
#include <mutex>

namespace El {
namespace Muxer {

namespace {

int GetBytesPerSample(PcmAacEncoder::PcmFormat format)
{
    switch (format) {
        case PcmAacEncoder::PcmFormat::S16LE:
            return 2;
        case PcmAacEncoder::PcmFormat::S32LE:
        case PcmAacEncoder::PcmFormat::FLT:
            return 4;
        default:
            return 2;
    }
}

std::once_flag g_gst_once_flag;

void InitGStreamer()
{
    EL_INFO("Initializing GStreamer...");
    gst_init(nullptr, nullptr);
    EL_INFO("GStreamer initialized successfully");
}

bool EnsureGStreamerInitialized()
{
    try {
        std::call_once(g_gst_once_flag, InitGStreamer);
    } catch (...) {
        EL_ERROR("Failed to initialize GStreamer runtime");
        return false;
    }
    return true;
}

} // namespace

PcmAacEncoder::PcmAacEncoder() = default;

PcmAacEncoder::~PcmAacEncoder()
{
    if (pipeline_) {
        gst_element_set_state(pipeline_, GST_STATE_NULL);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
    }
    appsrc_ = nullptr;
    appsink_ = nullptr;
}

const char *PcmAacEncoder::GetGstSampleFormat(PcmFormat format) const
{
    switch (format) {
        case PcmFormat::S16LE:
            return "S16LE";
        case PcmFormat::S32LE:
            return "S32LE";
        case PcmFormat::FLT:
            return "F32LE";
        default:
            return nullptr;
    }
}

GstCaps *PcmAacEncoder::CreateInputCaps(int sample_rate, int channels, PcmFormat format) const
{
    const char *fmt = GetGstSampleFormat(format);
    if (!fmt) {
        return nullptr;
    }

    return gst_caps_new_simple("audio/x-raw", "format", G_TYPE_STRING, fmt, "layout", G_TYPE_STRING, "interleaved",
                               "rate", G_TYPE_INT, sample_rate, "channels", G_TYPE_INT, channels, nullptr);
}

bool PcmAacEncoder::BuildPipeline(int sample_rate, int channels, PcmFormat pcm_format, int bitrate)
{
    const char *fmt = GetGstSampleFormat(pcm_format);
    if (!fmt) {
        EL_ERROR("Unsupported input PCM format");
        return false;
    }

    // 构建 GStreamer 管道字符串
    // 注意：appsrc 不设置 block，避免在初始化时阻塞状态转换
    char pipeline_desc[1024];
    snprintf(pipeline_desc, sizeof(pipeline_desc),
             "appsrc name=pcm_source format=time is-live=false "
             "! audio/x-raw,format=%s,layout=interleaved,rate=%d,channels=%d "
             "! audioconvert "
             "! audioresample "
             "! audio/x-raw,format=F32LE,rate=%d,channels=%d "
             "! avenc_aac bitrate=%d "
             "! aacparse "
             "! audio/mpeg,stream-format=adts "
             "! appsink name=aac_sink emit-signals=false sync=false drop=false max-buffers=32",
             fmt, input_sample_rate_, input_channels_, sample_rate, channels, bitrate);

    EL_INFO("Creating GStreamer pipeline: {}", pipeline_desc);

    GError *error = nullptr;
    pipeline_ = gst_parse_launch(pipeline_desc, &error);
    if (!pipeline_ || error) {
        EL_ERROR("Failed to create GStreamer pipeline: {}", error ? error->message : "unknown error");
        if (error) {
            g_error_free(error);
        }
        return false;
    }

    appsrc_ = gst_bin_get_by_name(GST_BIN(pipeline_), "pcm_source");
    appsink_ = gst_bin_get_by_name(GST_BIN(pipeline_), "aac_sink");

    if (!appsrc_ || !appsink_) {
        EL_ERROR("Failed to get appsrc or appsink from pipeline");
        return false;
    }

    // 设置 appsrc 的流类型和属性
    gst_app_src_set_stream_type(GST_APP_SRC(appsrc_), GST_APP_STREAM_TYPE_STREAM);

    // 不要在这里设置状态为 PLAYING，先设置为 PAUSED
    GstStateChangeReturn state_ret = gst_element_set_state(pipeline_, GST_STATE_PAUSED);
    if (state_ret == GST_STATE_CHANGE_FAILURE) {
        EL_ERROR("Failed to set GStreamer pipeline to PAUSED state");
        return false;
    }

    // 等待到达 PAUSED 状态
    GstState state = GST_STATE_VOID_PENDING;
    GstState pending = GST_STATE_VOID_PENDING;
    state_ret = gst_element_get_state(pipeline_, &state, &pending, 5 * GST_SECOND);

    EL_INFO("Pipeline state after PAUSED request: result={}, current={}, pending={}",
            static_cast<int>(state_ret), static_cast<int>(state), static_cast<int>(pending));

    if (state_ret == GST_STATE_CHANGE_FAILURE) {
        EL_ERROR("GStreamer pipeline state change to PAUSED failed");

        // 获取错误信息
        GstBus *bus = gst_element_get_bus(pipeline_);
        if (bus) {
            GstMessage *msg;
            while ((msg = gst_bus_pop(bus)) != nullptr) {
                GError *err = nullptr;
                gchar *debug_info = nullptr;

                switch (GST_MESSAGE_TYPE(msg)) {
                    case GST_MESSAGE_ERROR:
                        gst_message_parse_error(msg, &err, &debug_info);
                        EL_ERROR("GStreamer error from {}: {} (debug: {})",
                                GST_OBJECT_NAME(msg->src),
                                err->message,
                                debug_info ? debug_info : "none");
                        g_error_free(err);
                        g_free(debug_info);
                        break;
                    case GST_MESSAGE_WARNING:
                        gst_message_parse_warning(msg, &err, &debug_info);
                        EL_WARN("GStreamer warning from {}: {} (debug: {})",
                               GST_OBJECT_NAME(msg->src),
                               err->message,
                               debug_info ? debug_info : "none");
                        g_error_free(err);
                        g_free(debug_info);
                        break;
                    default:
                        break;
                }
                gst_message_unref(msg);
            }
            gst_object_unref(bus);
        }

        return false;
    }

    // 现在设置为 PLAYING - 由于已经在 PAUSED，这应该很快完成
    state_ret = gst_element_set_state(pipeline_, GST_STATE_PLAYING);
    if (state_ret == GST_STATE_CHANGE_FAILURE) {
        EL_ERROR("Failed to set GStreamer pipeline to PLAYING state");
        return false;
    }

    // 对于 appsrc，ASYNC 或 SUCCESS 都是正常的
    if (state_ret != GST_STATE_CHANGE_FAILURE) {
        EL_INFO("GStreamer pipeline set to PLAYING state (return: {})", static_cast<int>(state_ret));
    }

    return true;
}

bool PcmAacEncoder::Initialize(int sample_rate,
                               int channels,
                               PcmFormat pcm_format,
                               int bitrate,
                               int input_sample_rate,
                               int input_channels)
{
    if (initialized_) {
        EL_WARN("PcmAacEncoder already initialized");
        return true;
    }

    if (!EnsureGStreamerInitialized()) {
        return false;
    }

    if (input_sample_rate == 0) {
        input_sample_rate = sample_rate;
    }
    if (input_channels == 0) {
        input_channels = channels;
    }

    pcm_format_ = pcm_format;
    sample_rate_ = sample_rate;
    channels_ = channels;
    input_sample_rate_ = input_sample_rate;
    input_channels_ = input_channels;
    input_bytes_per_sample_ = GetBytesPerSample(pcm_format_);
    samples_pushed_ = 0;
    eos_pushed_ = false;
    encoder_flushed_ = false;

    if (pipeline_) {
        gst_element_set_state(pipeline_, GST_STATE_NULL);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
        appsrc_ = nullptr;
        appsink_ = nullptr;
    }

    if (!BuildPipeline(sample_rate_, channels_, pcm_format_, bitrate)) {
        if (pipeline_) {
            gst_element_set_state(pipeline_, GST_STATE_NULL);
            gst_object_unref(pipeline_);
        }
        pipeline_ = nullptr;
        appsrc_ = nullptr;
        appsink_ = nullptr;
        return false;
    }

    initialized_ = true;
    EL_INFO("PcmAacEncoder initialized using GStreamer (PCM -> AAC)");
    return true;
}

bool PcmAacEncoder::PushBuffer(const uint8_t *data, size_t size)
{
    if (!appsrc_ || !data || size == 0) {
        return true;
    }

    if (eos_pushed_) {
        EL_WARN("PushBuffer called after EOS was sent");
        return false;
    }

    if (size % (input_channels_ * input_bytes_per_sample_) != 0) {
        EL_ERROR("PCM data size {} is not aligned with sample format", size);
        return false;
    }

    GstBuffer *buffer = gst_buffer_new_allocate(nullptr, size, nullptr);
    if (!buffer) {
        EL_ERROR("Failed to allocate GstBuffer");
        return false;
    }

    GstMapInfo map;
    if (!gst_buffer_map(buffer, &map, GST_MAP_WRITE)) {
        EL_ERROR("Failed to map GstBuffer for writing");
        gst_buffer_unref(buffer);
        return false;
    }

    std::memcpy(map.data, data, size);
    gst_buffer_unmap(buffer, &map);

    int samples = static_cast<int>(size / (input_channels_ * input_bytes_per_sample_));
    GstClockTime pts = static_cast<GstClockTime>(samples_pushed_) * GST_SECOND / input_sample_rate_;
    GstClockTime duration = static_cast<GstClockTime>(samples) * GST_SECOND / input_sample_rate_;

    GST_BUFFER_PTS(buffer) = pts;
    GST_BUFFER_DURATION(buffer) = duration;

    samples_pushed_ += samples;

    GstFlowReturn flow = gst_app_src_push_buffer(GST_APP_SRC(appsrc_), buffer);
    if (flow != GST_FLOW_OK) {
        EL_ERROR("Failed to push buffer into appsrc (flow={})", static_cast<int>(flow));
        return false;
    }

    return true;
}

bool PcmAacEncoder::PullSamples(std::vector<uint8_t> &out, bool drain)
{
    if (!appsink_) {
        return false;
    }

    const GstClockTime timeout = drain ? GST_CLOCK_TIME_NONE : 0;

    while (true) {
        GstSample *sample = gst_app_sink_try_pull_sample(GST_APP_SINK(appsink_), timeout);
        if (!sample) {
            break;
        }

        GstBuffer *buffer = gst_sample_get_buffer(sample);
        if (buffer) {
            GstMapInfo map;
            if (gst_buffer_map(buffer, &map, GST_MAP_READ)) {
                out.insert(out.end(), map.data, map.data + map.size);
                gst_buffer_unmap(buffer, &map);
            } else {
                EL_ERROR("Failed to map GstBuffer for reading");
                gst_sample_unref(sample);
                return false;
            }
        }
        gst_sample_unref(sample);
    }

    if (drain && gst_app_sink_is_eos(GST_APP_SINK(appsink_))) {
        encoder_flushed_ = true;
    }

    return true;
}

bool PcmAacEncoder::ProcessInput(const uint8_t *data, size_t size)
{
    if (!initialized_) {
        EL_ERROR("PcmAacEncoder not initialized");
        return false;
    }

    if (encoder_flushed_) {
        EL_WARN("ProcessInput called after encoder was flushed");
        return false;
    }

    if (!PushBuffer(data, size)) {
        return false;
    }

    return true;
}

bool PcmAacEncoder::GetData(std::vector<uint8_t> &buffer, bool flush)
{
    buffer.clear();

    if (!initialized_) {
        EL_ERROR("PcmAacEncoder not initialized");
        return false;
    }

    if (flush && !encoder_flushed_ && !eos_pushed_) {
        GstFlowReturn flow = gst_app_src_end_of_stream(GST_APP_SRC(appsrc_));
        if (flow != GST_FLOW_OK) {
            EL_ERROR("Failed to send EOS to appsrc (flow={})", static_cast<int>(flow));
            return false;
        }
        eos_pushed_ = true;
    }

    if (!PullSamples(buffer, flush)) {
        return false;
    }

    if (flush && gst_app_sink_is_eos(GST_APP_SINK(appsink_))) {
        encoder_flushed_ = true;
    }

    return true;
}

bool PcmAacEncoder::Reset()
{
    if (!initialized_ || !pipeline_) {
        EL_ERROR("PcmAacEncoder reset requested but encoder not initialized");
        return false;
    }

    // Reset internal flags/timestamps
    encoder_flushed_ = false;
    eos_pushed_ = false;
    samples_pushed_ = 0;

    // Quickly clear EOS by state cycling: READY -> PAUSED -> PLAYING
    GstStateChangeReturn ret = gst_element_set_state(pipeline_, GST_STATE_READY);
    if (ret == GST_STATE_CHANGE_FAILURE) {
        EL_ERROR("Failed to set pipeline to READY during reset");
        return false;
    }

    GstState cur = GST_STATE_VOID_PENDING;
    GstState pend = GST_STATE_VOID_PENDING;
    gst_element_get_state(pipeline_, &cur, &pend, 2 * GST_SECOND);

    ret = gst_element_set_state(pipeline_, GST_STATE_PAUSED);
    if (ret == GST_STATE_CHANGE_FAILURE) {
        EL_ERROR("Failed to set pipeline to PAUSED during reset");
        return false;
    }
    gst_element_get_state(pipeline_, &cur, &pend, 2 * GST_SECOND);

    ret = gst_element_set_state(pipeline_, GST_STATE_PLAYING);
    if (ret == GST_STATE_CHANGE_FAILURE) {
        EL_ERROR("Failed to set pipeline to PLAYING during reset");
        return false;
    }

    EL_INFO("PcmAacEncoder fast reset complete");
    return true;
}

} // namespace Muxer
} // namespace El
